use crate::cus_executor_future::time_future::TimeFuture;
use crate::utils::ThreadSafePtr;
use crate::{safe_pin, thread_safe_ptr};
use lazy_static::lazy_static;
use log::info;
use std::arch::x86_64::_mm_or_ps;
use std::cell::RefCell;
use std::collections::HashMap;
use std::ops::Deref;
use std::pin::Pin;
use std::ptr::{addr_of, NonNull};
use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
use std::sync::{mpsc, Arc, Barrier, BarrierWaitResult, Mutex};
use std::thread;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
use tokio::io::AsyncSeek;
use tokio::pin;

// 实现一个简单线程池
// 定义JOB 闭包获取所有权，允许在线程间传递，保证生命周期
type Job = Box<dyn FnOnce() -> () + Send + 'static>;
type Message = Option<Job>;
lazy_static! {
    static ref DROP_POOL: Arc<Mutex<SimplePool>> = Arc::new(Mutex::new(SimplePool::new(100)));
}
// Receiver drop时有未处理完的数据会panic,自动回收,实现future特性,保证稳定性
pub struct AutoRec<R: Send + 'static> {
    rec: Option<Receiver<R>>,
}

impl<R: Send + 'static> AutoRec<R> {
    pub fn new(rec: Receiver<R>) -> AutoRec<R> {
        AutoRec { rec: Some(rec) }
    }
    pub fn get(mut self) -> R {
        let taken = self.rec.take().unwrap();
        taken.recv().unwrap()
    }
    // 回收
    pub(self) fn rev(rc: Receiver<R>) {
        DROP_POOL.lock().unwrap().drop_commit(move || {
            if let Err(e) = rc.recv_timeout(Duration::from_secs(30)) {
                eprintln!("{:?}", e);
            };
        });
    }
    pub(self) fn rev_all(rcs: Vec<Receiver<R>>) {
        DROP_POOL.lock().unwrap().drop_commit(move || {
            rcs.into_iter().for_each(|rc| {
                if let Err(e) = rc.recv_timeout(Duration::from_secs(30)) {
                    eprintln!("{:?}", e);
                };
            });
        });
    }

    pub(self) fn rev_all_sync(rcs: Vec<Receiver<R>>) {
        rcs.into_iter().for_each(|rc| {
            rc.recv().unwrap();
        });
    }
}

impl<R: Send + 'static> Drop for AutoRec<R> {
    fn drop(&mut self) {
        if let Some(taken) = self.rec.take() {
            AutoRec::rev(taken);
        };
    }
}

struct Worker {
    #[allow(unused)]
    id: usize,
    thread: Option<JoinHandle<()>>,
}

pub struct SimplePool {
    #[allow(unused)]
    max_cnt: usize,
    //最大线程数
    workers: Vec<Worker>,
    // 工作线程数组
    sender: mpsc::Sender<Message>,
    //通过mpsc实现pool对工作线程操控
    // 待处理缓冲队列()
    waiters: Option<Vec<Receiver<()>>>,
}

impl Worker {
    fn new(id: usize, rec: Arc<Mutex<Receiver<Message>>>) -> Self {
        Worker {
            id,
            thread: Some(thread::spawn(move || loop {
                let msg = rec.lock().unwrap().recv().unwrap(); // 不要直接match 会扩大锁所范围!!!
                match msg {
                    Some(job) => {
                        job();
                    }
                    None => break,
                }
            })),
        }
    }
}

impl SimplePool {
    // 异步回收废弃future
    pub(self) fn drop_commit<F>(&self, f: F)
    where
        F: FnOnce() -> () + Send + 'static,
    {
        self.sender.send(Some(Box::new(f))).unwrap();
    }

    pub fn new(max_cnt: usize) -> Self {
        assert!(max_cnt > 0, "max_cnt must bigger than zero!");
        let (sender, rc) = mpsc::channel();
        let mut workers = Vec::with_capacity(max_cnt);
        let rec = Arc::new(Mutex::new(rc));
        for id in 0..max_cnt {
            workers.push(Worker::new(id, Arc::clone(&rec)));
        }
        SimplePool {
            max_cnt,
            workers,
            sender,
            waiters: None,
        }
    }

    pub fn execute<F>(&self, f: F) -> AutoRec<()>
    where
        F: FnOnce() -> () + Send + 'static,
    {
        let (sd, rc) = mpsc::channel();
        self.sender
            .send(Some(Box::new(move || {
                sd.send(f()).unwrap();
            })))
            .unwrap();
        AutoRec::new(rc)
    }

    // 多个闭包必须使用动态分发处理,静态分发会认为是不同类型闭包
    pub fn execute_all(&self, fs: Vec<Box<dyn FnOnce() -> () + Send + 'static>>) {
        for f in fs {
            let (sd, rc) = mpsc::channel();
            self.sender
                .send(Some(Box::new(move || {
                    sd.send(f()).unwrap();
                })))
                .unwrap();
            AutoRec::rev(rc);
        }
    }

    pub fn commit<F, R>(&self, f: F) -> AutoRec<R>
    where
        F: FnOnce() -> R + Send + 'static,
        R: Send + 'static,
    {
        let (sd, rc) = mpsc::channel();
        self.sender
            .send(Some(Box::new(move || {
                sd.send(f()).unwrap();
            })))
            .unwrap();
        AutoRec::new(rc)
    }

    // 配合singleton使用,别面程序退出导致线程直接结束
    pub fn submit<F>(&mut self, f: F)
    where
        F: FnOnce() -> () + Send + 'static,
    {
        let (sd, rc) = mpsc::channel();
        self.sender
            .send(Some(Box::new(move || {
                sd.send(f()).expect("发送失败");
            })))
            .unwrap();
        match &mut self.waiters {
            None => {
                self.waiters = Some(vec![rc]);
            }
            Some(vec) => {
                vec.push(rc);
            }
        }
    }

    pub fn submit_all(&mut self, fs: Vec<Box<dyn FnOnce() -> () + Send + 'static>>) {
        for f in fs {
            let (sd, rc) = mpsc::channel();
            self.sender
                .send(Some(Box::new(move || {
                    sd.send(f()).expect("发送失败");
                })))
                .unwrap();
            match &mut self.waiters {
                None => {
                    self.waiters = Some(vec![rc]);
                }
                Some(vec) => {
                    vec.push(rc);
                }
            }
        }
    }

    pub fn flush(&mut self) {
        AutoRec::rev_all_sync(self.waiters.take().unwrap());
    }

    // #[deprecated]
    pub(self) fn join_all(&mut self) {
        for _ in self.workers.iter() {
            self.sender.send(None).unwrap();
        }
        self.workers.iter_mut().for_each(|worker| {
            if let Some(handle) = worker.thread.take() {
                handle.join().unwrap();
            };
        });
    }
}

#[allow(non_snake_case)]
#[macro_export]
macro_rules! execute_all {
   ($($func:expr),+)=>{
        crate::utils::thread_pool::SimplePool::single_const().execute_all(vec![$(std::boxed::Box::new(move || {$func})),*]);
    };

}

// single_create!(SimplePool, 64);

#[allow(non_snake_case)]
#[macro_export]
macro_rules! commit {
   ($($func:expr),*)=>{
        crate::utils::thread_pool::SimplePool::single_const().commit(move || {$($func),*})
    };

}

#[allow(non_snake_case)]
#[macro_export]
macro_rules! submit {
    ($($func:expr),*)=>{
        crate::utils::thread_pool::SimplePool::single_mut().lock().unwrap().submit(move || {$($func),*});
    };
}

#[allow(non_snake_case)]
#[macro_export]
macro_rules! submit_all {
   ($($func:expr),+)=>{
        crate::utils::thread_pool::SimplePool::single_mut().lock().unwrap().submit_all(vec![$(std::boxed::Box::new(move || {$func})),*]);
    };
}

#[allow(non_snake_case)]
#[macro_export]
macro_rules! flush_thread_pool {
    () => {
        crate::utils::thread_pool::SimplePool::single_mut()
            .lock()
            .unwrap()
            .flush();
    };
}

impl Drop for SimplePool {
    fn drop(&mut self) {
        self.join_all();
    }
}

#[allow(unused)]
pub fn time_cost_once<F: FnOnce() + 'static>(f: F) {
    let now = Instant::now();
    f();
    let duration = now.elapsed();
    println!("total time cost:");
    println!("cost {:?} ns", duration.as_nanos());
    println!("cost {:?} us", duration.as_micros());
    println!("cost {:?} ms", duration.as_millis());
    println!("cost {:?} ss", duration.as_secs());
}

#[allow(unused)]
pub fn time_cost_mut<F: FnMut()>(mut f: F) {
    let now = Instant::now();
    f();
    let duration = now.elapsed();
    println!("total time cost:");
    println!("cost {:?} ns", duration.as_nanos());
    println!("cost {:?} us", duration.as_micros());
    println!("cost {:?} ms", duration.as_millis());
    println!("cost {:?} ss", duration.as_secs());
}

#[allow(unused)]
pub fn time_cost<F: Fn()>(f: F) {
    let now = Instant::now();
    f();
    let duration = now.elapsed();
    println!("total time cost:");
    println!("cost {:?} ns", duration.as_nanos());
    println!("cost {:?} us", duration.as_micros());
    println!("cost {:?} ms", duration.as_millis());
    println!("cost {:?} ss", duration.as_secs());
}

// ///内部维护一个简单的测试器
// struct InnerTester {
//     test_name: String,
//     count: usize,
//     tester: Box<dyn FnOnce() -> () + Send + 'static>,
// }
///实现一个基本的批量多线程测试工具
pub struct MultiThreadTester {
    // 线程数
    thread_size: usize,
    //测试组
    test_map: HashMap<&'static str, (ThreadSafePtr<dyn Fn() + Sync + Send + 'static>, usize)>,
}

impl MultiThreadTester {
    pub fn new() -> Self {
        Self {
            thread_size: 0,
            test_map: HashMap::new(),
        }
    }
    pub fn append_test(
        &mut self,
        test_name: &'static str,
        tester: impl Fn() + Sync + Send + 'static,
        times: usize,
    ) -> &mut MultiThreadTester {
        self.thread_size += times;
        if self.test_map.contains_key(&test_name) {
            self.test_map.get_mut(&test_name).unwrap().1 += times;
        } else {
            self.test_map
                .insert(test_name, (ThreadSafePtr::capture(&tester), times));
        }
        self
    }
    pub fn run(self) {
        let mut vec = Vec::new();
        vec.reserve(self.thread_size);
        thread_safe_ptr! {
            let barrier = Barrier::new(self.thread_size);
        }
        let instant = Instant::now();
        self.test_map
            .into_iter()
            .for_each(|(test_name, (tester, times))| {
                for _ in 0..times {
                    vec.push(thread::spawn(move || {
                        barrier.wait();
                        tester();
                    }));
                }
            });

        for join_handle in vec {
            join_handle.join().expect("!!!tester finished failed!!!");
        }
        let duration = instant.elapsed();
        println!("MultiThreadTest time cost:");
        println!("cost {:?} ns", duration.as_nanos());
        println!("cost {:?} us", duration.as_micros());
        println!("cost {:?} ms", duration.as_millis());
        println!("cost {:?} ss", duration.as_secs());
    }
}

#[test]
fn do_test() {
    let mut tester = MultiThreadTester::new();
    tester
        .append_test(
            "test",
            || {
                println!("teee");
            },
            1000,
        )
        .append_test(
            "ok",
            || {
                println!("next");
            },
            10,
        );
    tester.run();
}
